package cn.itcast.flink.start;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 使用Flink 计算引擎实现流式数据处理：从Socket接收数据，实时进行词频统计WordCount
 *      TODO: 使用Flink提供工具类[ParameterTool]，获取传递参数
 */
public class _03FlinkWordCount {

	public static void main(String[] args) throws Exception {

		// TODO: 使用工具类，解析程序运行传递参数
		/*
			--host node1.itcast.cn --port 9999
		 */
		ParameterTool parameterTool = ParameterTool.fromArgs(args);
		if(parameterTool.getNumberOfParameters() != 2){
			System.out.println("Usage: _03FlinkWordCount --host node1.itcast.cn --port 9999");
			System.exit(-1);
		}
		final String host = parameterTool.get("host", "node1.itcast.cn") ;
		final Integer port = Integer.valueOf(parameterTool.get("port", "9999")) ;

		// 1.准备环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		//    2.准备数据-source
		DataStreamSource<String> inputDataStream = env.socketTextStream(host, port);

		//    3.处理数据-transformation
		// TODO: 流计算词频统计WordCount与处理思路基本一致
		SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = inputDataStream
			// step1. 分割单词
			.flatMap(new FlatMapFunction<String, String>() {
				@Override
				public void flatMap(String line, Collector<String> out) throws Exception {
					for (String word : line.trim().split("\\s+")) {
						out.collect(word);
					}
				}
			})
			// step2. 单词转换为二元组，表示出现一次
			.map(new MapFunction<String, Tuple2<String, Integer>>() {
				@Override
				public Tuple2<String, Integer> map(String word) throws Exception {
					return Tuple2.of(word, 1);
				}
			})
			// step3. 按照单词分组，进行求和
			.keyBy(0).sum(1);

		//    4.输出结果-sink
		resultDataStream.printToErr();

		// 5.触发执行-execute
		env.execute(_03FlinkWordCount.class.getSimpleName());
	}

}
